#include"Connection.h"
#include"TimeWheel.h"
#include"Parser.h"
#include"Config.h"
namespace wjc {
	const std::string Connection::GHOST = "ghost";

	Connection::Connection(Core* core, std::string name,int sockfd,
		const net::InetAddress& localAddr,const net::InetAddress& peerAddr,
		HeartType heartType):
		core_(core),name_(name), identity_(""),
		localAddr_(localAddr),
		peerAddr_(peerAddr),
		socket_(new net::Socket(sockfd)),
		channel_(new Channel(core, sockfd)),
		writerBuffer_(new Buffer(core->Id())),
		readerBuffer_(new Buffer(core->Id())),
		state_(kConnecting),
		heartType_(heartType) {
		timerHelper.start();
		if (heartType_== hSend) {
			heartChannel_=std::shared_ptr<TimerChannel>(new TimerChannel(core, Config::heartInterval));
			auto aliveMsg = Parser::getAliveMsg(this->getIdentity());
			heartChannel_->setOnTimerCallback(std::bind(&Connection::sendHeart, this, aliveMsg));
		}

		channel_->setReadCallback(std::bind(&Connection::handleRead, this, std::placeholders::_1));
		channel_->setWriteCallback(std::bind(&Connection::handleWriter, this));
		channel_->setCloseCallback(std::bind(&Connection::handleClose, this));
		channel_->setErrorCallback(std::bind(&Connection::handleError, this));

		socket_->setKeepAlive(true);
	}

	Connection::~Connection() {
		assert(channel_->isNoneEvent());
		//channel_->disableAll();
		//channel_->remove();
		timerHelper.end();
	
	}

	void Connection::handleRead(Timestamp receiveTime) {
		LOG_DEBUG << "handleRead enter1:" << TimeHelper::consumeTime();
		LOG_DEBUG << "handleRead enter1:" << TimeHelper::consumeTime();
		assert(core_->isInCoreThread());
		int error = 0;

		LOG_DEBUG << "readFromFd enter:" << TimeHelper::consumeTime();
		int n = readerBuffer_->readFromFd(channel_->fd(), &error);
		LOG_DEBUG << "readFromFd exit:" << TimeHelper::consumeTime();
		if(n==0)
			handleClose();
		else {
			if (n>0||(error == EINTR || error == EWOULDBLOCK || error == EAGAIN)) {
				if (messageCallback_)
					messageCallback_(shared_from_this(), readerBuffer_, receiveTime);
			}
		}
		LOG_DEBUG << "handleRead exit:" << TimeHelper::consumeTime();
	}
	void Connection::handleWriter() {
		LOG_DEBUG << "handleWriter enter:" << TimeHelper::consumeTime();
		assert(core_->isInCoreThread());
		int error; 
		writerBuffer_->writeToFd(channel_->fd(), &error);
		if (writerBuffer_->readableBytes() == 0) {
			channel_->disableWriting();
			if (writeCompleteCallback_)
				writeCompleteCallback_(shared_from_this());
		}
		LOG_DEBUG << "handleWriter exit:" << TimeHelper::consumeTime();
	}

	void Connection::handleClose() {
		assert(core_->isInCoreThread());
		setState(kDisconnected);
		channel_->disableAll();
		if (heartChannel_) heartChannel_->destruction();

		ConnectionPtr guardThis(shared_from_this());

		if (connectionCallback_)
			connectionCallback_(guardThis);
		if(closeCallback_)
			closeCallback_(guardThis);
	}
	void Connection::handleError() {
		assert(core_->isInCoreThread());
		int optval;
		socklen_t optlen = static_cast<socklen_t>(sizeof optval);

		if (::getsockopt(channel_->fd(), SOL_SOCKET, SO_ERROR, &optval, &optlen) < 0)
		{
			errno;
		}
		else
		{
			optval;
		}

	}

	void Connection::connectEstablished() {
		assert(core_->isInCoreThread());
		setState(kConnected);
		channel_->tie(shared_from_this());
		channel_->enableReading();
		if(connectionCallback_)
			connectionCallback_(shared_from_this());
		if (heartType_ != hNone) {
			std::shared_ptr<Entry> entryPtr(new Entry(shared_from_this()));
			core_->callAlive(entryPtr);
			entry_ = entryPtr;
		}
		if(heartChannel_)
			heartChannel_->start();
	}

	void Connection::connectDestroyed() {
		assert(core_->isInCoreThread());
		channel_->tie(shared_from_this());
		if (state_ == kConnected)
		{
			setState(kDisconnected);
			channel_->disableAll();
			if (heartChannel_) heartChannel_->destruction();
			if (connectionCallback_)
				connectionCallback_(shared_from_this());
		}
		channel_->remove();
	}
	void Connection::send(const std::string& str) {
		if (state_ != kConnected)return;
		if (core_->isInCoreThread()) {
			int old_len = writerBuffer_->writableBytes();
			if (writerBuffer_->write(str)) {
				if (!channel_->isWriting())
					channel_->enableWriting();
				if (highWaterMarkCallback_ && highWaterMark_ > 0) {
					if (old_len < highWaterMark_ && (old_len + str.length()) >= highWaterMark_)
						highWaterMarkCallback_(shared_from_this(), old_len + str.length());
				}
			}
		}
		else {
			core_->runInCore(CreateTask(std::bind(&Connection::send, shared_from_this(), str)));
		}
	}

	bool Connection::sendInCore(const char* begin, const char* end) {
		if (state_ != kConnected)return false;
		assert(core_->isInCoreThread());
		int old_len = writerBuffer_->writableBytes();
		if (writerBuffer_->write(begin, end-begin)) {
			if (!channel_->isWriting())
				channel_->enableWriting();
			if (highWaterMarkCallback_ && highWaterMark_ > 0) {
				if (old_len < highWaterMark_ && (old_len + end - begin) >= highWaterMark_)
					highWaterMarkCallback_(shared_from_this(), old_len + end - begin);
			}
		}
		return true;
	}

	void Connection::sendMsg(MessagePtr& msg) {
		if (core_->isInCoreThread()) { 			
			if (state_ != kConnected)return;
			int old_len = writerBuffer_->writableBytes();
			int write_len = 0;
			if (Parser::writeToBuffer(writerBuffer_.get(), *msg.get(), write_len)) {
				if (!channel_->isWriting())
					channel_->enableWriting();
				if (highWaterMarkCallback_ && highWaterMark_ > 0) {
					if (old_len < highWaterMark_ && (old_len + write_len) >= highWaterMark_)
						highWaterMarkCallback_(shared_from_this(), old_len + write_len);
				}
			}
		}
		else {
			core_->runInCore(CreateTask(std::bind(&Connection::sendMsg, shared_from_this(), msg)));
		}
	}

	void Connection::receive(std::promise<std::string>& pro) {
		std::future<std::string> fut;
		if (core_->isInCoreThread()) {
			std::string str = readerBuffer_->read();
			//printf("%s", str);
			if (!channel_->isReading())
				channel_->enableReading();
			pro.set_value(str);
		}
		else {
			core_->runInCore(CreateTask(std::bind(&Connection::receive, shared_from_this(), std::ref(pro))));
		}
	}

	bool Connection::receiveInCore(std::string& str) {
		if (core_->isInCoreThread()) {
			str = readerBuffer_->read();
			//printf("%s", str);
			if (!channel_->isReading())
				channel_->enableReading();
			return true;
		}
		return false;
	}
	int Connection::receiveInCore(char* begin, char* end) {
		assert(core_->isInCoreThread());
		return readerBuffer_->read(begin, end);
	}
	//void Connection::receive(std::string& str) {
	//	if (core_->isInCoreThread()) {
	//		str = readerBuffer_->read();
	//		printf("%s", str);
	//		if (!channel_->isReading())
	//			channel_->enableReading();
	//	}
	//	else {
	//		core_->runInCore(std::bind(&Connection::receive, shared_from_this(), str));
	//	}
	//}

	void Connection::close() {
		if (state_ == kConnected || state_ == kDisconnecting)
		{
			auto guard = shared_from_this();
			if (core_->isInCoreThread()) {
				setState(kDisconnected);
				handleClose();
			}
			else {
				core_->runInCore(CreateTask(std::bind(&Connection::close, guard)));
			}
		}
	}

	void Connection::sendHeart(MessagePtr& msg) {
		if (sendAliveCallback_)
			sendAliveCallback_(shared_from_this());
		sendMsg(msg);
	}

	void Connection::receiveHeart() {
		auto entry = entry_.lock();
		if (entry)
			core_->callAlive(entry);
		if (heartType_ == hRecv) {
			static auto msg = Parser::getAliveMsg(this->getIdentity());
			sendHeart(msg);
		}
	}


	void Connection::shutdown()
	{
		// FIXME: use compare and swap
		if (state_ == kConnected)
		{
			if (core_->isInCoreThread()) {
				if (!channel_->isWriting())
				{
					socket_->shutdownWrite();
				}
			}
			else {
				setState(kDisconnecting);
				// FIXME: shared_from_this()?
				core_->runInCore(CreateTask(std::bind(&Connection::shutdown, shared_from_this())));
			}
		}
	}

	//???
	void Connection::expired() {
		shutdown();
	}
}